/*
 * I write in my free time, for personal using
 * when i fell proud of it, i will give it to public
 * now,have nothing more to say
 */
package com.quangtin3.tunll;

import com.quangtin3.trans.TransReceiver;
import com.quangtin3.trans.TransSender;
import com.quangtin3.msg.Msg;
import com.quangtin3.msg.Channel;
import com.quangtin3.utils.Sequence;
import com.quangtin3.utils.ControlAble;
import com.quangtin3.utils.TunllException;
import com.quangtin3.utils.TunllLogger;
import com.quangtin3.utils.Utils;

import java.util.Hashtable;
import java.util.LinkedList;

/**
 * call it: center process unit :D
 * process all data from tunll, forward data message to AgentSender and perform 
 * command message
 * @author quanghx
 */
public class Processor implements Runnable, ControlAble {

    private Channel channel;
    private LinkedList<Msg> jobs = new LinkedList<Msg>();
    private TransReceiver transReceiver;
    private TransSender transSender;
    private Hashtable<Integer, AgentSender> agentSenders =
            new Hashtable<Integer, AgentSender>();
    private Hashtable<Integer, AgentReceiver> agentReceivers =
            new Hashtable<Integer, AgentReceiver>();

    public Processor(Channel channel) {
        this.channel = channel;

        transReceiver = new TransReceiver(this, channel);
        transSender = new TransSender(channel);
    }

    public void run() {
        try {
            TunllLogger.sys.info("thread started");
            //start Trans threads
            new Thread(transReceiver).start();
            new Thread(transSender).start();
            TunllLogger.sys.info("create transReceiver and transSender thread");

            //process data in this trans
            while (true) {
                //get job out of queue
                synchronized (jobs) {
                    while (jobs.isEmpty()) {
                        jobs.wait();
                    }
                }

                Msg job = jobs.remove(0);
                TunllLogger.sys.debug("process " + job.toString());

                //process this job
                switch (job.getType()) {
                    case Msg.TYPE_DATA:
                        //queue to AgentSender
                        if (agentSenders.containsKey(job.getTunnel())) {
                            agentSenders.get(job.getTunnel()).queue(job);
                        }
                        else {
                            TunllLogger.sys.error("data to tunnel that not exist!");
                        }
                        break;
                    case Msg.TYPE_COMAND:
                        performCmd(job);
                        break;
                    case Msg.TYPE_RESPONSE:
                        TunllLogger.sys.debug("we don't care about response now");
                        break;
                    default:
                        TunllLogger.sys.error("something misbehavior");
                        break;
                }
            }
        }
        catch (InterruptedException e) {
            TunllLogger.sys.error("interrupted " + e.getMessage());
        }
    }

    private void performCmd(Msg job) {
        switch (job.getCmd()) {
            case Msg.CMD_CLOSE:
                TunllLogger.sys.debug("closing tunnel " + job.getTunnel());
                if (agentSenders.containsKey(job.getTunnel()) && agentReceivers.containsKey(job.getTunnel())) {
                    agentSenders.remove(job.getTunnel()).requestStop();
                    agentReceivers.remove(job.getTunnel()).requestStop();
                }
                else {
                    TunllLogger.sys.debug("this tunnel has some isus");
                }
                break;
            case Msg.CMD_CONNECT:
                TunllLogger.sys.debug("create tunnel " + job.getTunnel());
                try {
                    String add = new String(job.getData());
                    String host = Utils.getHost(add);
                    int port = Utils.getPort(add);
                    int tunnel = Sequence.giveMeId();

                    Channel agentChannel = new Channel(host, port);
                    channel.connect();

                    //add to hasttable
                    agentReceivers.put(tunnel, new AgentReceiver(agentChannel, tunnel, transSender));
                    agentSenders.put(tunnel, new AgentSender(agentChannel));

                    //start them
                    new Thread(agentSenders.get(tunnel)).start();
                    new Thread(agentReceivers.get(tunnel)).start();
                }
                catch (TunllException e) {
                    TunllLogger.sys.error("create tunnel fail " + e.getMessage());
                }
                break;
            default:
                TunllLogger.sys.error("something misbehavior " + job.toString());
                break;
        }
    }

    /**
     * add job to my queue
     * @param job
     */
    public void queue(Msg job) {
        synchronized (jobs) {
            TunllLogger.sys.debug("queue " + job.toString());
            jobs.addLast(job);
            jobs.notify();
        }
    }

    /**
     * close channel associated with this object and
     * stop thread running it. release all resource
     */
    public void requestStop() {

    }

    public void requestSuspend() {

    }

    public void requestResume() {

    }

    public boolean isAlive() {
        return false;
    }

    /**
     * start process, need call only one.
     * if internal thread is stopped, what happend?
     */
    public synchronized void requestStart() {
    }
}
